# Author: tang
# Date: 2024-05-17
# Description: 新架构采集程序
# Previous version: /

# 标准库
import time
import json
import logging
import datetime
import socket
import traceback
import threading

import redis
# import pymysql

# 自建库
import collect_func as cf
import public_func as pf

from log import logger_log
# from log import logger_log

# logging.basicConfig(format='%(asctime)s-%(filename)s-[line:%(lineno)d]-%(levelname)s: %(message)s', level=logging.DEBUG)
# logging.basicConfig(format='%(asctime)s-%(filename)s-[line:%(lineno)d]-%(levelname)s: %(message)s', level=logging.ERROR,
#                     filename='collect.log')

# sqlite_conn = pf.sqlite_connect()
# sqlite_cursor = sqlite_conn.cursor()
# sqlite_cursor.execute('select * from communication_settings')
# comm_data = sqlite_cursor.fetchone()
# columns_tuple = sqlite_cursor.description
# columns_list = [field_tuple[0] for field_tuple in columns_tuple]
# logging.debug(columns_list)
# logging.debug(
#     comm_data)  # (1, 'tcp_server', '192.168.1.60', 31000, '{"register_whitelist":["192.168.1.239"]}', '127.0.0.1', 6379, 'gzdq.123', '127.0.0.1', 5672, 'admin', 'admin', 'routingkeyData', 0, '127.0.0.1', 1883, 'gzdq', 'admin', 'admin', 0, 'serial_server')
# comm_dict = dict(zip(columns_list, comm_data))
# logging.debug(
#     comm_dict)  # {'id': 1, 'communication_type': 'tcp_server', 'communication_ip': '192.168.1.60', 'communication_port': 31000, 'register_whitelist': '{"register_whitelist":["192.168.1.239"]}', 'redis_host': '127.0.0.1', 'redis_port': 6379, 'redis_pwd': 'gzdq.123', 'rabbitmq_host': '127.0.0.1', 'rabbitmq_port': 5672, 'rabbitmq_username': 'admin', 'rabbitmq_pwd': 'admin', 'rabbitmq_routingkey': 'routingkeyData', 'is_transmit_rabbitmq': 0, 'mqtt_host': '127.0.0.1', 'mqtt_port': 1883, 'mqtt_clientid': 'gzdq', 'mqtt_username': 'admin', 'mqtt_pwd': 'admin', 'is_transmit_mqtt': 0, 'connect_device': 'serial_server'}

# sqlite_cursor.execute('select * from collector_settings where enable=1')
# coll_data = sqlite_cursor.fetchall()
# columns_tuple = sqlite_cursor.description
# columns_list = [field_tuple[0] for field_tuple in columns_tuple]
# logging.debug(columns_list)
# logging.debug(
#     coll_data)  # [(38, 'defaultName', '192.168.1.239_30002', '{"collector_config":[{"device_code":"gz0124010002","slave_id":2,"device_type":"01","temp_numb":0},{"device_code":"gz0124010003","slave_id":1,"device_type":"01","temp_numb":0}]}', 600, 720, 0, '', 1), (39, 'defaultName', '192.168.1.239_30001', '{"collector_config":[{"device_code":"gz0124010001","slave_id":1,"device_type":"01","temp_numb":0}]}', 600, 720, 0, '', 1)]
# coll_list = []
# for i in coll_data:
#     data_dict = dict(zip(columns_list, i))
#     logging.debug(data_dict)
#     coll_list.append(data_dict)
# logging.debug(
#     coll_list)  # [{'id': 38, 'bus_name': 'defaultName', 'bus_register_id': '192.168.1.239_30002', 'collector_config': '{"collector_config":[{"device_code":"gz0124010002","slave_id":2,"device_type":"01","temp_numb":0},{"device_code":"gz0124010003","slave_id":1,"device_type":"01","temp_numb":0}]}', 'acqu_interval': 600, 'acqu_overtime': 720, 'controller_board': 0, 'bus_parameter1': '', 'enable': 1}, {'id': 39, 'bus_name': 'defaultName', 'bus_register_id': '192.168.1.239_30001', 'collector_config': '{"collector_config":[{"device_code":"gz0124010001","slave_id":1,"device_type":"01","temp_numb":0}]}', 'acqu_interval': 600, 'acqu_overtime': 720, 'controller_board': 0, 'bus_parameter1': '', 'enable': 1}]
# sqlite_conn.close()  # 关闭数据库连接

# communication_type = comm_dict['communication_type']
# if communication_type == 'tcp_server':  # TCP服务端
#     ip = comm_dict['communication_ip']
#     port = comm_dict['communication_port']
# elif communication_type == 'tcp_client':  # TCP客户端
#     ip = ''
#     port = ''
# elif communication_type == 'serial':  # 串口方式
#     ip = ''
#     port = ''
# else:
# ip = '192.168.0.122'
# ip = '0.0.0.0'
# port = '6666'

# coll_dict = {}
redis_conn = pf.redis_connect()
# redis_conn = redis.StrictRedis(host='localhost', port=6379, db=0, password='gzdq.123')  # todo test

# for i in coll_list:  # redis注册
#     coll_dict[i['bus_register_id']] = i

#     collector_config = json.loads(i['collector_config'])
#     collector_config = collector_config['collector_config']
#     logging.debug(collector_config)
#     reg_1 = {"bus_name": i["bus_name"], "bus_register_id": i["bus_register_id"],
#             "connect_device": comm_dict["connect_device"],
#             "connect_type": comm_dict["communication_type"], "ip_address": comm_dict["communication_ip"],
#             "port": comm_dict["communication_port"],
#             "collector_config": collector_config}
#     reg_dict = {"data_type": "register", "data": json.dumps(reg_1),
#                 "update_time": datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')}
#     reg_json = json.dumps(reg_dict)
#     redis_conn.rpush(pf.REDIS_CACHE_KEY, reg_json)
#     logging.info('[%s]:redis set register data success' % i['bus_register_id'])
# logging.debug(
#     coll_dict) 
    # {'192.168.1.239_30002': {'id': 38, 'bus_name': 'defaultName', 'bus_register_id': '192.168.1.239_30002', 'collector_config': 
    # '{"collector_config":[{"device_code":"gz0124010002","slave_id":2,"device_type":"01","temp_numb":0},
    # {"device_code":"gz0124010003","slave_id":1,"device_type":"01","temp_numb":0}]}', 
    # 'acqu_interval': 600, 'acqu_overtime': 720, 'controller_board': 0, 'bus_parameter1': '', 'enable': 1}, 
    # '192.168.1.239_30001': {'id': 39, 'bus_name': 'defaultName', 'bus_register_id': '192.168.1.239_30001', 'collector_config': 
    # '{"collector_config":[{"device_code":"gz0124010001","slave_id":1,"device_type":"01","temp_numb":0}]}',
    #  'acqu_interval': 600, 'acqu_overtime': 720, 'controller_board': 0, 'bus_parameter1': '', 'enable': 1}}

def register_device_channel(comm_dict,coll_dict):
    register_id_str=comm_dict["communication_ip"]+'_'+str(comm_dict["communication_port"])
    if not coll_dict:
        return
    collector_config = coll_dict[register_id_str]['collector_config']
    logging.debug(collector_config)
    reg_1 = { "bus_register_id": register_id_str,
            "connect_device": comm_dict["connect_device"],
            # "connect_type": comm_dict["communication_type"], 
            "ip_address": comm_dict["communication_ip"],
            "port": comm_dict["communication_port"],
            "collector_config": collector_config}
    reg_dict = {"data_type": "register", "data": json.dumps(reg_1),
                "update_time": datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')}
    reg_json = json.dumps(reg_dict)
    redis_conn.rpush(pf.REDIS_CACHE_KEY, reg_json)
    logging.info('[%s]:redis set register data success' % register_id_str)
    logging.debug(
        coll_dict)  



def get_bus_communication_info(ip,port):
    """获取总线通信信息"""
    _comm_dict,_coll_dict={},{}
    
    mysql_conn = pf.mysql_connect()  # mysql连接
    with mysql_conn.cursor() as cursor:
        sql_comm = ''' select br.ip_address,br.`port`,br.connect_device,br.`status`,br.collect_interval,br.control_timeout,br.control_status 
                    from bus_register br inner join system_buswhitelist bwl 
                    on br.white_list_id=bwl.id where br.`status`=1 and br.`port`='%s' and br.ip_address='%s' limit 1;'''%(str(port),str(ip))
        cursor.execute(sql_comm)
        sel_comm_data = cursor.fetchone()
        if sel_comm_data:
            _comm_dict['communication_ip'] = sel_comm_data[0]
            _comm_dict['communication_port'] = sel_comm_data[1]
            _comm_dict['connect_device'] = sel_comm_data[2]
            _comm_dict['status'] = sel_comm_data[3]
            _comm_dict['collect_interval'] = sel_comm_data[4]
            _comm_dict['control_timeout'] = sel_comm_data[5]
            _comm_dict['control_status'] = sel_comm_data[6]
        else:
            return _comm_dict,_coll_dict
            # _comm_list.append(_comm_dict)
        sql_coll =''' select dr.device_code,dr.slave_id,dr.device_type,dr.temp_numb,bus.ip_address,bus.`port`,dr.dept_id from device_register dr 
                inner join (select br.id,br.ip_address,br.`port` from bus_register br 
                inner join system_buswhitelist bwl on br.white_list_id=bwl.id  where br.`status`=1 and br.`port`='%s' and br.ip_address='%s') bus 
                on dr.fk_bus_register_id = bus.id;'''%(str(port),str(ip))
        cursor.execute(sql_coll)
        sel_coll_datas = cursor.fetchall()
        if sel_coll_datas:
            _coll_list=[]
            for sel_coll_data in sel_coll_datas:
                _coll_dict1={
                    'device_code':sel_coll_data[0],
                    'slave_id':sel_coll_data[1],
                    'device_type': sel_coll_data[2],
                    'temp_numb':sel_coll_data[3],
                    'dept_id':sel_coll_data[6]
                }
                _coll_list.append(_coll_dict1)
            _coll_dict={
                sel_coll_data[4]+'_'+str(sel_coll_data[5]):
                {
                "collector_config":_coll_list
                }}
    mysql_conn.close()
    logging.debug(_comm_dict)
    logging.debug(_coll_dict)
    register_device_channel(_comm_dict,_coll_dict)
    return _comm_dict,_coll_dict



    
class ColService(threading.Thread):
    """采集服务"""

    def __init__(self, ip, port, redis_conn):
        super().__init__()
        self.ip = ip  # 服务端IP
        self.port = int(port)  # 服务端端口
        # self.coll_dict = coll_dict  # 总线信息
        self.redis_conn = redis_conn  # redis连接
        self.threads=[]
        self.sock=[]
    def run(self):
        logging.info('--ColService start--')
        ser_sk = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        ser_sk.setsockopt(socket.SOL_SOCKET, socket.SO_KEEPALIVE, 1)  # 开启长连接
        ser_sk.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)  # 端口复用
        ser_sk.bind((self.ip, self.port))
        ser_sk.listen(128)
        while True:
            try:
                cli_sk, cli_addr = ser_sk.accept()
                cli_sk.setblocking(True)  # 阻塞
                cli_sk.settimeout(5)
                self.sock.append(cli_sk)
                """获取总线通信信息"""
                comm_dict,coll_dict=get_bus_communication_info(cli_addr[0],cli_addr[1]) # 获取总线通信信息
                if not self._validate_connection(comm_dict, coll_dict, cli_sk, cli_addr):
                    continue
                register_id = ''
                """处理连接"""
                register_id = self._get_register_id(comm_dict, cli_sk, cli_addr)
                if not register_id:
                    continue
                logging.info('[ColService]-Accept new connection from %s' % register_id)
                # logging.info('[ColService]-Accept new connection from %s:%s' % cli_addr)

                if comm_dict['status'] == 1:
                    """启动采集线程"""
                    self._start_collect_thread(cli_sk, register_id, coll_dict, comm_dict)
                else:
                    time.sleep(5)
                logging.debug('---------------')

            except Exception as e:
                logging.error(traceback.format_exc())

    def handle_collect(self, cli_sk, register_id, collector_list, acqu_interval, acqu_overtime, controller_board,
                       bus_parameter1):
        # device_data = json.loads(collector_config)
        # device_list = device_data['collector_config']
        device_list = collector_list
        logging.debug(device_list)

        dev_para_list = []    # 该列表存储已读取过参数的设备
        sock_status = True    # socket状态（True为正常，False为异常）
        sleep_status = False  # 休眠状态（True为需要休眠，False为不休眠）
        while True:
            acqu_interval ,bisstate= self.interval_query(register_id,cli_sk)
            if bisstate == False:
                break
            for i in device_list:
                try:
                    device_code = i['device_code']
                    slave_id = i['slave_id']
                    device_type = i['device_type']
                    temp_numb = i.get('temp_numb', 3)
                    logging.debug((device_code, slave_id, device_type, temp_numb))  # ('gz0124010003', 1, '01', 0)
                    if device_type == '01':  # 开关柜八通道
                        cf.eight_channel_collect(self.redis_conn, cli_sk, device_code, slave_id, device_type, temp_numb,
                                                 dev_para_list)
                    elif device_type == '02':  # 电缆三通道
                        cf.three_channel_collect(self.redis_conn, cli_sk, device_code, slave_id, device_type,
                                                 dev_para_list)
                    elif device_type == '03':  # 站房式局放（暂无）
                        pass
                    elif device_type == '04':  # 485版本超声地电二合一
                        cf.aa_tev_collect(self.redis_conn, cli_sk, device_code, slave_id, device_type, dev_para_list)
                    elif device_type == '05':  # LORA版本超声地电二合一
                        pass
                    elif device_type == '06':  # 429三合一
                        cf.aa_tev_uhf_collect(self.redis_conn, cli_sk, device_code, slave_id, device_type, dev_para_list)
                    elif device_type == '07':  # 带电指示器
                        pass
                    elif device_type == '08':  # 八声八电(AA+HF)
                        pass
                    elif device_type == '09':  # 配网低功耗
                        pass
                    elif device_type == '10':  # todo 无线测温
                        cf.wireless_temperature_collect(self.redis_conn, cli_sk, device_code, slave_id, device_type,
                                                        temp_numb, dev_para_list)
                    elif device_type == '11':  # 机械特性（暂无）
                        pass
                    elif device_type == '12':  # todo 接地、护层环流
                        cf.circulation_collect(self.redis_conn, cli_sk, device_code, slave_id, device_type,
                                               dev_para_list)
                    elif device_type == '13':  # 光纤测温（暂无）
                        pass
                    elif device_type == '14':  # 电缆隧道综合监测_边代（暂无）
                        pass
                    elif device_type == '15':  # 变电站多状态智能监测系统（暂无）
                        pass
                    elif device_type == '16':  # 近距离多状态智能监测系统（暂无）
                        pass
                    elif device_type == '27':  # 南网voltage+tev+temp+water
                        cf.voltage_tev_temp_water_collect(self.redis_conn, cli_sk, device_code, slave_id, device_type)
                        time.sleep(5)
                        sleep_status = True
                    elif device_type == '80':  # 中盛测温
                        cf.ntc_temperature_collect(self.redis_conn, cli_sk, device_code, slave_id,
                                                                     device_type, temp_numb)
                        time.sleep(5)
                        sleep_status = True
                    if device_code not in dev_para_list:
                        dev_para_list.append(device_code)
                except redis.RedisError as e:
                    logging.error(traceback.format_exc())
                    self.redis_conn.close()
                    logging.info('redis_conn closed')
                    time.sleep(60)  # 每隔60s重连一次
                    self.redis_conn = pf.redis_connect()
                    logging.info('redis reconnect')
                except socket.error as e:
                    logging.error(traceback.format_exc())
                    sock_status = False
                except Exception as e:
                    logging.error(traceback.format_exc())

            if sleep_status or not sock_status:
                try:
                    cli_sk.close()
                    logging.info('[%s]:socket closed' % register_id)
                except Exception as e:
                    logging.error(traceback.format_exc())
                break

            time.sleep(acqu_interval)
            # time.sleep(20)  # todo test
    def interval_query(self,register_id,cli_sk):
        bus=register_id.split('_')
        comm_dict,coll_dict=get_bus_communication_info(bus[0],bus[1])
        if not comm_dict or not coll_dict:
            #关闭线程采集
            for index, bus_info in enumerate(self.threads):
                t = bus_info['bus_thread']
                rid = bus_info['register_id']
                _cli_sk = bus_info['client']
                # cli_addr = bus_info['cli_addr']
                if rid == register_id and _cli_sk==cli_sk:
                    if cli_sk in self.sock:
                        cli_sk.close() #强制断开
                        self.sock.remove(cli_sk)
                    if t.is_alive() and t is not threading.current_thread():
                        
                        bus_info['stop_event'].set()
                        try:
                            t.join(timeout=2)
                        except Exception as e:
                            logging.error('[%s]: 等待线程退出时发生异常: %s', register_id, e)
                        # del self.threads[index]
                        logging.info('[%s]:close thread' % register_id)
                        # self.disconnect_client(cli_addr)
                        self.threads.pop(index)
                        return 0,False
            return 0,False
        bus_interval= comm_dict.get('collect_interval',600)
        bus_status= comm_dict.get('status',0)
        if bus_status==0:
            #关闭线程采集
            for index, bus_info in enumerate(self.threads):
                t = bus_info['bus_thread']
                rid = bus_info['register_id']
                _cli_sk = bus_info['client']
                # cli_addr = bus_info['cli_addr']
                if rid == register_id and _cli_sk==cli_sk:
                    if cli_sk in self.sock:
                        cli_sk.close() #强制断开
                        self.sock.remove(cli_sk)
                    if t.is_alive() and t is not threading.current_thread():
                       
                        bus_info['stop_event'].set()
                        try:
                            t.join(timeout=2)
                        except Exception as e:
                            logging.error('[%s]: 等待线程退出时发生异常: %s', register_id, e)
                  
                        # del self.threads[index]
                        logging.info('[%s]:close thread' % register_id)
                        self.threads.pop(index)
                        
                        return 0,False
        return bus_interval, True
    def _validate_connection(self, comm_dict, coll_dict, cli_sk, cli_addr):
        """验证连接是否有效"""
        if not comm_dict:
            self._close_and_remove_socket(cli_sk)
            time.sleep(5)
            return False

        if not coll_dict:
            self._close_and_remove_socket(cli_sk)
            time.sleep(5)
            return False

        return True
    def _close_and_remove_socket(self, cli_sk):
        """关闭并移除Socket"""
        if cli_sk in self.sock:
            cli_sk.close()
            self.sock.remove(cli_sk)
    def _get_register_id(self, comm_dict, cli_sk, cli_addr):
        """获取设备的注册ID"""
        connect_device = comm_dict.get('connect_device', '')
        if connect_device == 'serial_server':  # 串口服务器
            return cli_addr[0] + '_' + str(cli_addr[1])
        elif connect_device == 'dtu':  # 4G DTU
            try:
                rcv_bytes = cli_sk.recv(1024)
                logging.debug(rcv_bytes)
                rcv_str = rcv_bytes.decode()  # ASCII
                if rcv_str.startswith('3A') and rcv_str.endswith('0D0A'):
                    return rcv_str[2:-4]
            except socket.error as e:
                logging.error(traceback.format_exc())
                cli_sk.close()
                return None
        return None
    def _start_collect_thread(self, cli_sk, register_id, coll_dict, comm_dict):
        """启动采集线程"""
        collector_list = coll_dict[register_id]['collector_config']
        acqu_interval = comm_dict['collect_interval']
        acqu_overtime = comm_dict['control_timeout']
        controller_board = comm_dict['control_status']
        bus_parameter1 = comm_dict.get('bus_parameter1', '')

        logging.debug((collector_list, type(collector_list)))
        logging.debug((acqu_interval, type(acqu_interval)))
        logging.debug((acqu_overtime, type(acqu_overtime)))
        logging.debug((controller_board, type(controller_board)))
        logging.debug((bus_parameter1, type(bus_parameter1)))

        stop_event = threading.Event()
        t = threading.Thread(target=self.handle_collect, args=(
            cli_sk, register_id, collector_list, acqu_interval, acqu_overtime, controller_board, bus_parameter1))
        bus_info = {"bus_thread": t, "stop_event": stop_event, 'register_id': register_id, "client": cli_sk}
        self.threads.append(bus_info)
        t.start()



if __name__ == '__main__':
    try:
        col_thread = ColService('0.0.0.0', str(pf.BUSS_CONFIG['local_port']),  redis_conn)
        col_thread.start()
    except Exception as e:
        logging.error(traceback.format_exc())
